package com.hex.ds.hdrs.agent.service;

import com.hex.ds.hdrs.agent.dao.AgentDao;
import com.hex.ds.hdrs.agent.po.AgentConf;
import com.hex.ds.hdrs.app.po.AppConf;
import com.hex.ds.hdrs.app.service.AppManager;
import com.hex.ds.hdrs.center.po.CenterConf;
import com.hex.ds.hdrs.common.constant.Const;
import com.hex.ds.hdrs.common.constant.ErrorCode;
import com.hex.ds.hdrs.common.exception.BizException;
import com.hex.ds.hdrs.common.utils.BaseUtil;
import com.hex.ds.hdrs.common.utils.HdrsDateUtil;
import com.hex.ds.hdrs.cron.po.CronTask;
import com.hex.ds.hdrs.log.service.LogService;
import com.hex.ds.hdrs.period.task.dto.ResultTask;
import com.hex.ds.hdrs.period.task.dto.ScheduleTask;
import com.hex.ds.hdrs.period.task.dto.TaskInfo;
import com.hex.ds.hdrs.period.task.po.PeriodTask;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Package: com.hex.ds.hdrs.engine.po
 * @ClassName AgentSession
 * @Description: Agent 内存信息存储
 * @Author: jd.chen
 * @Date 2021/7/5 17:56
 * @Version v1.0
 **/
@Slf4j
@Data
public class AgentSession {

    // periodRunningList 同步锁
    private final byte[] periodListSyn = {};
    // cronRunningList 同步锁
    private final byte[] cronListSyn = {};

    //是否在资源重新计算中
    private String mark = Const.AGENT_MARK_STOP;

    //注册成功时间
    private String registerTs;
    //心跳时间
    private long lstHeartBeatTs;

    // 节点编码
    private String agentCode;
    // 节点配置
    private AgentConf agentConf;
    // 节点管理
    private AgentManager agentManager;
    // 节点状态
    private String status;
    // token
    private String token;
    // 当前占用资源数
    private final AtomicInteger curResNum = new AtomicInteger(0);
    // 当前占用并发数
    private final AtomicInteger curConcurrentNum = new AtomicInteger(0);

    // 应用管理器
    private AppManager appManager;
    // 应用配置列表
    private List<AppConf> appConfList = new ArrayList<>();
    // 应用数
    private int appSize = 0;

    // 正在运行队列(周期)
    private List<PeriodTask> periodRunningList = Collections.synchronizedList(new ArrayList<>());
    // 正在运行队列(定时)
    private List<CronTask> cronRunningList = Collections.synchronizedList(new ArrayList<>());
    // 重新计算资源时临时运行队列(周期)
    private List<PeriodTask> deltaList = Collections.synchronizedList(new ArrayList<>());

    // 发送失败重试次数
    private int disSendFailLimit;
    // 重发间隔
    private int disSendSleep;

    // 调用agent接口
    private AgentDao agentDao;
    private LogService logService;

    public AgentSession(AgentConf agentConf, List<AppConf> appConfList, AgentManager agentManager) {
        this.registerTs = HdrsDateUtil.getCurTs();
        this.lstHeartBeatTs = HdrsDateUtil.getMillis(new Date());
        this.agentCode = agentConf.getAgentCode();
        this.agentConf = agentConf;
        this.agentManager = agentManager;
        this.status = agentConf.getAgentStatus();
        this.token = this.createToken();
        this.appManager = agentManager.getAppManager();
        this.appConfList = appConfList;
        this.appSize = this.appConfList.size();
        this.agentDao = agentManager.getAgentDao();
        this.logService = agentManager.getLogService();
        this.disSendFailLimit = agentManager.getDisSendFailLimit();
        this.disSendSleep = agentManager.getDisSendSleep();

        for (AppConf appConf : appConfList) {
            if (null != appManager.getAppSession(appConf.getAppCode())) {
                appManager.getAppSession(appConf.getAppCode()).addAgentSession(this);
            }
        }
        agentManager.addAgentSession(this);
    }

    /**
     * 创建token
     *
     * @return
     */
    private String createToken() {
        String token = BaseUtil.getPkId();
        log.info("【AgentSession.createToken】======token: {}", token);
        return token;
    }

    /**
     * 验证是否有效的agentsession
     *
     * @param task
     * @return
     */
    public boolean checkRes(PeriodTask task) {
        log.debug("验证是否有效的agentsession【AgentSession.checkRes】=======》");

        // 任务资源数
        int taskResNum = task.getResNum();
        // 获取当前节点资源数
        int limitResNum = agentConf.getResNum() - taskResNum;
        // 判断资源数是否满足
        if (this.getCurResNum().get() > limitResNum) {
            log.debug("【节点{}资源数不足】【checkRes】taskResNum:{},curResNum:{},limitRes:{}", agentCode, taskResNum, curResNum.get(), agentConf.getResNum());
            //logService.resNotEnoughAgentLog(agentConf, "节点资源数达到最大!");
            return false;
        }

        // 获取当前节点并发数
        int limitConcurrentNum = agentConf.getConcurrentNum() - 1;
        // 判断并发数是否满足
        if (this.getCurConcurrentNum().get() > limitConcurrentNum) {
            log.debug("【节点{}并发数不足】【checkConcurrent】curConcurrentNum:{},limitConcurrentNum: {}", agentCode, curConcurrentNum.get(), agentConf.getConcurrentNum());
            //logService.resNotEnoughAgentLog(agentConf, "节点并发数达到最大!");
            return false;
        }

        return true;
    }

    /**
     * 发送至节点
     *
     * @param task
     * @return
     */
    @SneakyThrows
    public void sendToAgent(PeriodTask task) {
        log.info("发送至节点【sendToAgent】taskCode:{},bizDate: {}", task.getTaskCode(), task.getBizDate());
        for (int i = 0; i < disSendFailLimit; i++) {
            if (this.sendPeriodTask(task)) {
                return;
            } else {
                // 界面配置的任务出错 再次发送间隔时间
                Thread.sleep(Integer.parseInt(task.getErrDifferDuration()) * 1000L);
            }
        }
        throw new BizException(ErrorCode.ERROR_820005);
    }

    /**
     * 发送至节点
     *
     * @param cronTask
     * @return
     */
    @SneakyThrows
    public void sendCronToAgent(CronTask cronTask) {
        log.info("定时任务发送至节点【sendCronToAgent】taskCode: {}", cronTask.getTaskCode());
        for (int i = 0; i < disSendFailLimit; i++) {
            if (this.sendCronTask(cronTask)) {
                return;
            } else {
                // 界面配置的任务出错 再次发送间隔时间
                Thread.sleep(disSendSleep);
            }
        }
        throw new BizException(ErrorCode.ERROR_820005);
    }

    /**
     * 发送到节点
     *
     * @param task
     */
    public boolean sendPeriodTask(PeriodTask task) {
        log.debug("【任务发送失败处理】taskCode:{},bizDate: {}", task.getTaskCode(), task.getBizDate());
        TaskInfo taskInfo = taskToTaskInfo(task);
        return agentDao.sendPeriodTask(taskInfo, agentConf);
    }

    /**
     * 定时任务发送到节点
     *
     * @param task
     */
    public boolean sendCronTask(CronTask task) {
        TaskInfo taskInfo = cronToTaskInfo(task);
        return agentDao.sendCronTask(taskInfo, agentConf);
    }

    /**
     * 查看agent的running中任务
     *
     * @param task
     */
    public String qryAgentRunning(PeriodTask task) {
        TaskInfo taskInfo = taskToTaskInfo(task);
        return agentDao.qryAgentRunning(taskInfo, agentConf);
    }


    /**
     * task对象转TaskInfo
     *
     * @param task
     * @return
     */
    private TaskInfo taskToTaskInfo(PeriodTask task) {
        ScheduleTask scheduleTask = new ScheduleTask();
        scheduleTask.setAgentCode(task.getAgentCode());
        scheduleTask.setBatchJobCode(task.getBatchTag());
        scheduleTask.setBizDate(task.getBizDate());
        scheduleTask.setExecEndTs(task.getExecEndTs());
        scheduleTask.setExecStTs(task.getExecStTs());
        scheduleTask.setExtParam(task.getExtParam());
        scheduleTask.setTaskName(task.getTaskName());
        scheduleTask.setResNum(task.getResNum());
        scheduleTask.setSendTime(HdrsDateUtil.getCurrentTime());
        scheduleTask.setShellPath(task.getShellPath());
        scheduleTask.setTaskCode(task.getTaskCode());
        scheduleTask.setTaskStatus(task.getTaskStatus());
        scheduleTask.setTaskType("CYCLE");
        scheduleTask.setJobCode(task.getJobCode());
        scheduleTask.setOverPlanDuration(task.getOverPlanDuration());
        scheduleTask.setAppCode(task.getAppCode());
        TaskInfo taskInfo = new TaskInfo();
        ResultTask resultTask = new ResultTask();
        taskInfo.setResultTask(resultTask);
        taskInfo.setTask(scheduleTask);
        return taskInfo;
    }

    /**
     * task对象转TaskInfo
     *
     * @param task
     * @return
     */
    private TaskInfo cronToTaskInfo(CronTask task) {
        ScheduleTask scheduleTask = new ScheduleTask();
        scheduleTask.setPkId(task.getPkId());
        scheduleTask.setAgentCode(task.getAgentCode());
        scheduleTask.setBatchJobCode(task.getBatchTag());
        scheduleTask.setExecEndTs(task.getExecEndTs());
        scheduleTask.setExecStTs(task.getExecStTs());
        scheduleTask.setExtParam(task.getExtParam());
        scheduleTask.setTaskName(task.getTaskName());
        scheduleTask.setResNum(task.getResNum());
        scheduleTask.setSendTime(HdrsDateUtil.getCurrentTime());
        scheduleTask.setShellPath(task.getShellPath());
        scheduleTask.setTaskCode(task.getTaskCode());
        scheduleTask.setTaskStatus(task.getTaskStatus());
        scheduleTask.setTaskType("CRON");
        scheduleTask.setAppCode(task.getAppCode());
        TaskInfo taskInfo = new TaskInfo();
        ResultTask resultTask = new ResultTask();
        taskInfo.setResultTask(resultTask);
        taskInfo.setTask(scheduleTask);
        return taskInfo;
    }

    /**
     * 更新session 周期任务变化
     *
     * @param task
     */
    public void addTaskSession(PeriodTask task) {
        log.info("更新session 周期任务变化【AgentSession.addTaskSession】开始=======》");
        synchronized (periodListSyn) {
            if (this.getCurConcurrentNum().get() + 1 > this.getAgentConf().getConcurrentNum()) {
                throw new BizException(ErrorCode.ERROR_820003);
            }
            List<PeriodTask> taskList = new ArrayList<>();
            taskList.add(task);
            if (this.periodRunningList.addAll(taskList)) {
                this.curResNum.addAndGet(task.getResNum());
                this.curConcurrentNum.addAndGet(1);
            }
        }
    }

    /**
     * 更新session 删除周期任务
     *
     * @param task
     */
    public void delTaskSession(PeriodTask task) {
        log.info("更新session 删除周期任务【AgentSession.delTaskSession】开始=======》");
        try {
            //判断是否在资源重新计算中
            if (this.mark.equals(Const.AGENT_MARK_STOP)) {
                synchronized (periodListSyn) {
                    List<PeriodTask> taskList = new ArrayList<>();
                    taskList.add(task);
                    if (this.periodRunningList.removeAll(taskList)) {
                        this.curResNum.addAndGet(-task.getResNum());
                        this.curConcurrentNum.addAndGet(-1);
                    } else {
                        log.error("【delTaskSession】删除失败=========={}====={}", task.getTaskCode(), task.getBizDate());
                    }
                }
            } else {
                List<PeriodTask> taskList = new ArrayList<>();
                taskList.add(task);
                this.periodRunningList.removeAll(taskList);
                this.deltaList.addAll(taskList);
            }
        } catch (Exception e) {
            log.error("【delTaskSession】异常:{}", e.getMessage(), e);
        }
    }

    /**
     * 更新session 周期任务变化
     *
     * @param taskCode
     */
    public void delTaskSession(String taskCode) {
        log.info("更新session 删除周期任务【AgentSession.delTaskSession】taskCode: {}", taskCode);
        try {
            for (int i = 0; i < this.periodRunningList.size(); i++) {
                PeriodTask periodTask = this.periodRunningList.get(i);
                if (periodTask != null && taskCode.equals(periodTask.getTaskCode())) {
                    this.delTaskSession(periodTask);
                    return;
                }
            }
        } catch (Exception e) {
            log.error("【{}】【delTaskSession】异常:", taskCode, e);
        }
    }

    /**
     * 更新session 删除定时任务
     *
     * @param cronTask
     */
    public void delTaskSession(CronTask cronTask) {
        log.info("更新session 删除定时任务【AgentSession.delTaskSession】开始=======》");
        synchronized (cronListSyn) {
            List<CronTask> taskList = new ArrayList<>();
            taskList.add(cronTask);
            this.cronRunningList.removeAll(taskList);
        }
    }

    /**
     * 更新最新心跳时间
     */
    public void updateHeartBeatTs() {
        this.lstHeartBeatTs = HdrsDateUtil.getMillis(new Date());
    }

    /**
     * 测试agent连接
     *
     * @return
     */
    public boolean testLink() {
        return agentDao.testLink(agentConf);
    }

    /**
     * 通知agent新的master初始化完成
     *
     * @param newCenterConf
     */
    public void masterInitSucc(CenterConf newCenterConf) {
        try {
            boolean res = agentDao.masterInitSucc(agentConf, newCenterConf.getCenterIp());
            if (!res) {
                log.warn("agent未接收到master初始化成功【masterInitSucc】agentCode: {}", agentCode);
            }
        } catch (Exception e) {
            log.error("【masterInitSucc】异常:{}", e.getMessage(), e);
        }
    }

    /**
     * 杀死agent上任务进程
     *
     * @param taskInfo
     * @return
     */
    public boolean killTask(TaskInfo taskInfo) {
        try {
            return agentDao.killTask(agentConf, taskInfo);
        } catch (Exception e) {
            log.error("【killTask】异常:{}", e.getMessage(), e);
            return false;
        }
    }

    public String getAgentName() {
        return agentConf.getAgentName();
    }

    public boolean containTask(PeriodTask task) {
        return periodRunningList.contains(task);
    }

    /**
     * @Method: againAgentResAndCurr <br>
     * @Param: [] <br>
     * @Return: void <br>
     * @Description：重新计算agentSession资源数和并发数<br>
     * @Author： gc.zuo<br>
     * @Date： 2021/6/30 9:19 <br>
     * @Version： V1.0<br>
     */
    public void againAgentResAndCurr() {
        log.info("重新计算agentSession资源数和并发数");
        try {
            synchronized (periodListSyn) {
                List<PeriodTask> taskList = this.periodRunningList;
                int resNumTotal = 0;
                //遍历计算资源数
                for (PeriodTask task : taskList) {
                    resNumTotal += task.getResNum();
                }
                //重新赋值
                int deltaResNumTotal = 0;
                for (PeriodTask task : this.deltaList) {
                    deltaResNumTotal += task.getResNum();
                }
                this.curResNum.set(resNumTotal - deltaResNumTotal);
                this.curConcurrentNum.set(taskList.size() - this.deltaList.size());
                this.deltaList.clear();
            }
        } catch (Exception e) {
            log.error("{}重新计算资源数并发数出错:{}", this.agentCode, e.getMessage(), e);
        }
    }
}
